package cn.doitedu.rtdw.dash_board;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/9/30
 * @Desc: 学大数据，上多易教育
 **/
public class Job4_累计订单额 {
    public static void main(String[] args) {
        // 创建编程入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        TableConfig config = tenv.getConfig();
        config.getConfiguration().setBoolean("table.exec.emit.early-fire.enabled", true);
        config.getConfiguration().setString("table.exec.emit.early-fire.delay", "5000 ms");


        // 建cdc连接器表，映射mysql中的 订单表
        // 1. 建表映射 业务库中的  oms_order表
        tenv.executeSql(
                "CREATE TABLE order_mysql (    " +
                        " id            INT ,             " +
                        " status        INT ,             " +
                        " total_amount  decimal(10,0) ,   " +
                        " pay_amount    decimal(10,0) ,   " +
                        " create_time   timestamp(3)  ,   " +
                        " payment_time  timestamp(3)  ,   " +
                        " delivery_time timestamp(3)  ,   " +
                        " confirm_time  timestamp(3)  ,   " +
                        " modify_time   timestamp(3)  ,   " +
                        " rt as create_time ,             " +
                        " watermark for rt as rt - interval '0' second ,   " +
                        " PRIMARY KEY (id) NOT ENFORCED            " +
                        " ) WITH (                                 " +
                        "     'connector' = 'mysql-cdc',               " +
                        "     'hostname' = 'doitedu'   ,               " +
                        "     'port' = '3306'          ,               " +
                        "     'username' = 'root'      ,               " +
                        "     'password' = 'root'      ,               " +
                        "     'database-name' = 'realtimedw' ,         " +
                        "     'table-name' = 'oms_order'               " +
                        ")"
        );

        // 今天累计到当前的gmv总额，应付(或实付)总额，优惠总额（要求每5分钟更新一次）


        /**
         * 用TVF窗口聚合底层的 changelog 流，会报如下错误
         * StreamPhysicalWindowAggregate doesn't support consuming update and delete changes which is produced by node TableSourceScan(table=[[default_catalog, default_database, order_mysql]], fields=[id, status, total_amount, pay_amount, create_time, payment_time, delivery_time, confirm_time, modify_time])
         *
         * 基于TVF语法的窗口聚合，不支持消费 -U/+U/-D 语义（讲白了就是不支持消费changelog流，只支持appendOnly流）
         *
         * GroupByWindow语法（已经被flink官方标记为 deprecated），则支持消费changelog流
         */
//        tenv.executeSql(
//                " select                                                                                             "+
//                        "      window_start,                                                                         "+
//                        " 	   window_end,                                                                         "+
//                        "      sum(total_amount) as total_amount,                                                    "+
//                        " 	   sum(pay_amount) as pay_amount,                                                      "+
//                        " 	   sum(total_amount - pay_amount ) as discount_amount                                  "+
//                        "                                                                                            "+
//                        " from table(                                                                                "+
//                        "     cumulate(table order_mysql,descriptor(rt), interval '5' minute ,interval '24' hour)    "+
//                        " )                                                                                          "+
//                        " group by                                                                                   "+
//                        "   window_start,                                                                            "+
//                        " 	window_end                                                                              "
//        ).print();


        /**
         * 使用GroupByWindow和cumulate窗口来聚合底层的changelog流，会报如下错误：
         * Exception in thread "main" org.apache.flink.table.api.ValidationException:
         *  SQL validation failed. Cannot call table function here: 'CUMULATE'
         *
         *  GroupByWindow聚合，不支持 cumulate窗口，只支持  tumble()窗口 和 hop()窗口
         */
//        tenv.executeSql(
//                "select                                                                        "+
//                        " cumulate_start(rt,interval '5' minute ,interval '24' hour) as window_start, "+
//                        "	cumulate_end(rt,interval '5' minute ,interval '24' hour) as window_end,     "+
//                        " sum(total_amount) as total_amount,                                          "+
//                        "	sum(pay_amount) as pay_amount,                                              "+
//                        "	sum(total_amount - pay_amount ) as discount_amount                          "+
//                        "from order_mysql                                                               "+
//                        "group by                                                                       "+
//                        "   cumulate(rt,interval '5' minute ,interval '24' hour)                        "
//        ).print();


        /**
         * 用 GroupByWindow  +  tumble窗口  + early-fire 窗口触发器
         * 来实现 TVF语法中的  cumulate窗口效果
         */
        tenv.executeSql(
                "select                                                                        " +
                        "   tumble_start(rt,interval '24' hour) as window_start,                        "+
                        "   sum(total_amount) as total_amount,                                          " +
                        "	sum(pay_amount) as pay_amount,                                              " +
                        "	sum(total_amount - pay_amount ) as discount_amount                          " +
                        "from order_mysql                                                               " +
                        "group by                                                                       " +
                        "   tumble(rt,interval '24' hour)                                               "
        ).print();

    }
}
